package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author:ray
 * @time:2022/10/15 14:16
 * @description:
 **/
public class DwdInteractionComment {
    public static void main(String[] args) {
        //TODO 1、环境配置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2、检查点设置
        //2.1 设置检查点
        /*env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 job 取消之后检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");*/
        //TODO 3、从kafka的topic_db主题中读取数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_interaction_review_group"));

        //TODO 4、筛选出评论表
        /*{"database":"edu","table":"review_info","type":"insert","ts":1665715969,"xid":69718,"commit":true,"data":{"id":3675,"user_id":950,"course_id":292,"review_txt":null,"review_stars":5,"create_time":"2022-10-14 10:52:49","deleted":"0"}}*/
        Table reviewInfo = tableEnv.sqlQuery(
                "select " +
                        " data['id'] id, " +
                        " data['user_id'] user_id, " +
                        " data['course_id'] course_id, " +
                        " data['review_txt'] review_txt, " +
                        " data['review_stars'] review_stars, " +
                        " data['create_time'] create_time, " +
                        " data['deleted'] deleted," +
                        "ts " +
                        "from topic_db " +
                        "where `table`='review_info' and `type`='insert'"
        );
        //TODO 5、创建dwd_interaction_comment表，向其中插入数据
        tableEnv.executeSql("" +
                "create table dwd_interaction_comment(" +
                " id string, " +
                " user_id string, " +
                " course_id string, " +
                " review_txt string, " +
                " review_stars string, " +
                " create_time string, " +
                " deleted string, " +
                " ts string, " +
                " primary key(id) not enforced ) "
                +MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));
        reviewInfo.executeInsert("dwd_interaction_comment");
    }
}
